package com.neusoft.databus.core.hbase;

import java.io.IOException;

import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.security.UserGroupInformation;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.hadoop.config.annotation.EnableHadoop;
import org.springframework.data.hadoop.config.annotation.SpringHadoopConfigurerAdapter;
import org.springframework.data.hadoop.config.annotation.builders.HadoopConfigConfigurer;
import org.springframework.data.hadoop.hbase.HbaseTemplate;

import lombok.extern.slf4j.Slf4j;

/**
 * description
 *
 * @author sunchf
 * @date 2019年3月28日 下午1:28:56
 */
@Slf4j
@Configuration
@ConditionalOnClass(HbaseTemplate.class)
@ConditionalOnProperty(name = "databus.core.hbase.enabled", havingValue = "true")
@EnableHadoop
public class HBaseConfiguration extends SpringHadoopConfigurerAdapter {

    @Autowired
    private org.apache.hadoop.conf.Configuration configuration;

    @Bean
    public HBaseProperties hbaseProperties() {
        return new HBaseProperties();
    }

    @Override
    public void configure(final HadoopConfigConfigurer config) throws Exception {
        config.withResources().resources(this.hbaseProperties().getResources());
        if (this.hbaseProperties().getSecurity().isEnabled()) {
            System.setProperty("java.security.krb5.conf", this.hbaseProperties().getSecurity().getConf());
            config.withSecurity().authMethod(this.hbaseProperties().getSecurity().getAuthMethod());
            config.withSecurity().userPrincipal(this.hbaseProperties().getSecurity().getUserPrincipal());
            config.withSecurity().userKeytab(this.hbaseProperties().getSecurity().getUserKeytab());
        }
    }

    @Bean
    public HbaseTemplate hbaseTemplate(final org.apache.hadoop.conf.Configuration configuration) {
        final HbaseTemplate hbaseTemplate = new HbaseTemplate(configuration);
        hbaseTemplate.setAutoFlush(true);
        return hbaseTemplate;
    }

    @Bean
    public HBaseAdmin hbaseAdmin() {
        try {
            if (this.hbaseProperties().getSecurity().isEnabled()) {
                UserGroupInformation.setConfiguration(this.configuration);
                UserGroupInformation.loginUserFromKeytab(this.hbaseProperties().getSecurity().getUserPrincipal(),
                        this.hbaseProperties().getSecurity().getUserKeytab());
            }

            final Connection conn = ConnectionFactory.createConnection(this.configuration);
            final Admin admin = conn.getAdmin();
            return (HBaseAdmin) admin;
        } catch (final IOException e) {
            HBaseConfiguration.log.error(e.getMessage(), e);
        }

        return null;
    }
}
